Description

The parallel apply system enables VReplication to apply binlog events using multiple concurrent MySQL connections instead of a single serial connection. When --vreplication-parallel-replication-workers is set to N > 1, incoming transactions are analyzed for writeset conflicts and dispatched to N worker goroutines, each with its own MySQL connection. Transactions that touch different primary keys run concurrently; transactions that conflict are serialized.

The design is inspired by MySQL's own multi-threaded replica applier (MTA), specifically the WRITESET dependency tracking mode. We recompute writesets on the target side from the row events themselves, rather than relying on the source's binlog_transaction_dependency_tracking setting. This means the parallel applier works regardless of whether the source uses COMMIT_ORDER or WRITESET dependency tracking — or no dependency tracking at all. That is a critical property as we need to support importing databases of various versions and configurations into Vitess.

The design — most notably the in order commits of transactions (see more below) — also supports another critical property which is that it must live alongside the traditional serial applier/vplayer and provide the same external/visible semantics (the position and lag management, metrics, etc). This will be experimental for some time and it needs to be given time to bake before we can even consider fully replacing the serial applier code. We cannot risk destabilizing this critical component or primitive within VReplication because performance without correctness is worthless.

Goroutine Architecture

Four types of goroutines cooperate in a pipeline:

VStream → relayLog → [scheduleLoop] → applyScheduler → [workerLoop ×N] → commitCh → [commitLoop]
                          │                                                              │
                          └──── vp.serialMu (main connection) ───────────────────────────┘

1. scheduleLoop (single goroutine)

The schedule loop reads batched events from the relay log, parses them into logical transactions, builds writesets, and enqueues applyTxn structs into the scheduler. It runs on the main goroutine of applyEventsParallel.

Responsibilities:

• Fetches event batches from the relay log via relay.Fetch()
• Tracks transaction boundaries (GTID → events → COMMIT)
• Extracts commit metadata (sequenceNumber, commitParent) from GTID events
• Classifies transactions as row-only or mixed (DDL, OTHER, JOURNAL events make a transaction non-row-only; FIELD events are metadata and do not affect classification)
• Builds PK-based writeset keys for row-only transactions
• Handles transaction batching (merging consecutive transactions)
• Handles empty transactions via unsavedEvent (bypasses the scheduler entirely)
• Handles heartbeats inline (not enqueued through the scheduler)
• Handles throttling by estimating lag and updating time_throttled

2. workerLoop (N goroutines)

Each worker goroutine calls scheduler.nextReady() to block until a transaction is ready to execute, then applies its row events using the worker's private MySQL connection.

Responsibilities:

• Waits for the scheduler to declare a transaction ready
• Builds a narrow worker-local view of the vplayer (workerLocalVPlayer) once per worker lifetime, exposing only the fields workers may share (tablePlans, replicatorPlan, serialMu, etc); foreign_key_checks session state is tracked per connection on the vdbClient, initialized when the worker's connections are created
• Applies each event via worker.applyEvent(), which rebinds the worker's dbClient/query/commit onto that local view (the orchestrator's vplayer is never mutated)
• On completion, attaches the worker's connection info to the payload and sends the transaction to commitCh
• Blocks on txn.done until commitLoop finishes committing (this prevents the worker from reusing its connection while commitLoop is still writing to _vt.vreplication on it)
• Worker connections run at READ COMMITTED: the writeset model covers PK/unique/FK conflicts but cannot see InnoDB gap/next-key locks, which REPEATABLE READ takes even for point operations on absent rows (e.g. a DELETE of a row that doesn't exist, or delete-marking in a non-unique secondary index). A later-ordered transaction's gap lock blocking an earlier-ordered transaction's INSERT would deadlock through the commitLoop's strict ordering — invisible to InnoDB's deadlock detector (MySQL's MTA has its Commit_order_manager for exactly this). RC takes no gap locks for row-image application and is MySQL's own recommendation for row-based parallel replicas; statement-based events force-serialize, so RC cannot change their outcome.

3. commitLoop (single goroutine)

The commit loop receives completed transactions from workers and commits them in strict order (by order field). This guarantees that the position saved to _vt.vreplication is monotonically increasing.

Responsibilities:

• Maintains a pending map and nextOrder counter
• Buffers out-of-order arrivals; commits in strict sequence
• For worker transactions: writes the position update and COMMIT directly on the worker's connection (via the payload's client/query/commit) WITHOUT holding serialMu, so slow MySQL commits never block the scheduleLoop; only the brief vplayer bookkeeping afterwards takes serialMu
• For commitOnly transactions (DDL, OTHER, JOURNAL, position saves): applies events and updates position on the main connection
• Updates lag metrics after each commit
• Signals the scheduler via markCommitted() to release inflight state
• Returns io.EOF when stop position is reached

4. applyScheduler (shared state, not a goroutine)

The scheduler is a shared data structure protected by a mutex. It determines which transactions can execute concurrently based on writeset conflicts and transaction classification.

Transaction Classification

Every transaction enqueued to the scheduler is classified with these boolean flags:

Ready-Check Hierarchy (isReadyLocked)

The scheduler checks readiness in this order:

1. noConflict → always ready. These are position-only saves that have no data and no side effects beyond updating _vt.vreplication. They bypass all conflict checking to prevent deadlocks where a position save with an earlier order is blocked by inflight data transactions.

2. inflightGlobal > 0 → blocked. Any inflight global transaction blocks everything.

3. forceGlobal → ready only when ALL inflight counters are zero (no inflight transactions of any kind).

4. hasCommitMeta with inflightMissingMeta > 0 → blocked. Transactions with commit metadata cannot run alongside transactions lacking it (safety boundary).

5. hasCommitMeta with non-empty writeset → ready if no inflight writeset conflicts. This is the key optimization: writeset-only conflict detection, skipping the commit-parent dependency check entirely. This allows parallelism even when the source is not using WRITESET based dependency tracking, which would otherwise produce a strict serial chain.

6. hasCommitMeta with empty writeset → falls back to commit-parent ordering (commitParent <= lastCommittedSequence).

7. No commit metadata, no writeset → treated as global (increments inflightGlobal).

8. No commit metadata, has writeset → checks writeset conflicts + must wait if inflightCommitMeta > 0.

Inflight Tracking

Four counters track what's currently being applied:

Writeset Conflict Detection

PK-Based Keys

For each row change (INSERT, UPDATE, DELETE), the writeset extractor hashes the table name and primary key values into a uint64 using xxhash (XXH64). Conceptually the key represents tableName:pk1,pk2,..., but using fixed-size hashes instead of heap-allocated strings eliminates the dominant per-transaction allocation source at high TPS. Both the before-image and after-image are hashed because an UPDATE that changes a PK value must conflict with both the old and new key.

The PK column indices are stored in TablePlan.PKIndices (a []bool where true at index i means column i is part of the PK). This is populated when the replicator plan is built from FIELD events.

Unique-Key Keys

Unique secondary indexes make transactions on different rows order-dependent: one transaction frees a unique value and another claims it, so PK keys alone would schedule the pair in parallel and the second to apply would hit a duplicate-key error. Mirroring MySQL's WRITESET dependency tracking — which hashes every unique key, not just the PK — the writeset extractor also emits a key per hashable unique secondary index for both row images, with the index ordinal folded into the digest so different indexes on the same table occupy distinct key spaces. A NULL in any key column emits no key for that image (MySQL unique indexes permit multiple NULLs, so a NULL-valued key cannot conflict with anything). Only transactions actually colliding on a unique value serialize against each other; everything else stays parallel.

FK-Aware Keys

When foreign key constraints exist, a transaction that modifies a child table row must serialize with transactions that modify the referenced parent row. At startup, the parallel applier queries information_schema.KEY_COLUMN_USAGE to discover all FK constraints. For each child row change, it hashes the parent table name and FK column values into a uint64 — using the same xxhash scheme as PK keys, so the hash will match the parent table's PK-based writeset hash. This forces the scheduler to see a conflict between child and parent operations.

Copy-on-Write Table Plan Snapshot

The scheduleLoop needs to read tablePlans to build writesets, but the serial applier path mutates tablePlans when FIELD events arrive. To avoid holding a read lock across writeset computation, we use a copy-on-write snapshot: snapshotTablePlans copies the map only when tablePlansVersion has changed since the last snapshot. FIELD events increment the version.

Serialization Escapes (fail closed)

Whenever the writeset hasher cannot prove the absence of a conflict, the transaction is routed to the serial path (forceGlobal) rather than guessed at:

• Unhashable unique indexes: prefix-length and expression/functional unique indexes enforce uniqueness over a derived value the hasher cannot reproduce (a full-value hash would miss conflicts), and a PK that doesn't match the replication identity breaks the keys-imply-identity reasoning — tables carrying these force-serialize. Plain-column unique secondaries are handled by unique-key hashing (see Unique-Key Keys above) instead of serialization.
• Unsupported column mappings: plans that rewrite, project, or reorder columns (expressions, generated columns, lossy casts) produce hash inputs that don't correspond 1:1 to the row image, so they serialize.
• Partial row images (DataColumns/JsonPartialValues, noblob -1 lengths on relevant columns): omitted columns can shift values into wrong field slots, so they serialize.
• No usable identity: a plan with no PK/identity columns contributes an error (not silently zero keys) so the txn serializes instead of racing untracked.
• DDL: after an executed DDL, FK metadata is re-queried from information_schema and a per-table stale-plan barrier force-serializes transactions touching affected tables until their refreshed FIELD event replaces the cached plan.

Transaction Batching

The schedule loop merges consecutive transactions in the same relay log fetch into a single larger transaction, mirroring the serial applier's hasAnotherCommit lookahead. This reduces the number of MySQL COMMITs.

Batching rules:

• If another COMMIT exists ahead in the current relay batch and we don't need to force-save, skip the flush and let events accumulate
• Time-bounded at 500ms: during catch-up, heartbeats may not arrive to force a flush, so we force one every 500ms to keep time_updated and lag metrics fresh
• Skipped when FK refs exist: large batches merge many parent/child operations into a single writeset, causing nearly all batches to conflict on FK ref keys and effectively serializing everything. Flushing each source transaction individually lets the scheduler detect truly independent transactions.

When batching merges a transaction, its sequenceNumber is advanced via scheduler.advanceCommittedSequence() so that future transactions whose commitParent references the merged-away sequence are not blocked forever.

Empty Transaction Handling

Empty transactions (GTID → BEGIN → COMMIT with no row events) are very common (from filtered-out tables on the same source shard). They bypass the scheduler entirely:

• The COMMIT is stored as vp.unsavedEvent
• If idleTimeout (1 second) passes with no real commits, the position is saved as a noConflict commit-only transaction via enqueueCommitOnly
• The scheduler's lastCommittedSequence is advanced for the empty transaction's sequence number, preventing dependent transactions from being blocked

During catch-up, empty transactions can arrive in a continuous stream, preventing the idle timeout from firing. A separate lastHeartbeatRefresh timer periodically updates time_updated directly via SQL to keep max_v_replication_lag fresh.

Commit Ordering

Commits must be strictly ordered because _vt.vreplication stores a single position, and that position must only move forward. The commitLoop achieves this with:

1. Each transaction gets a monotonically increasing order number (assigned by scheduleLoop)
2. The commitLoop maintains nextOrder starting at 1
3. Transactions arrive out of order via commitCh; they're buffered in a pending map
4. Transactions are committed strictly in order: nextOrder is incremented after each commit
5. After committing, markCommitted() releases the transaction's inflight state in the scheduler, potentially unblocking waiting transactions (a released multi-key writeset can ready several pending transactions; dispatched workers pass the wakeup baton so all of them start without waiting for the next commit)

End-to-end backpressure: the scheduler caps outstanding ordered work at roughly one applying transaction per worker plus the commit buffer (maxOutstandingOrders ≈ 5× workers), so a commitLoop stalled on an early order bounds how far the pipeline can run ahead of durable progress.

Worker Transaction Commit Protocol

When committing a worker's transaction, the commitLoop (without holding serialMu for any of the MySQL work):

1. Writes the position update on the worker's connection via updatePosWithoutStop — inside the worker's open MySQL transaction (in batch mode this rides the same multi-statement flush as the COMMIT)
2. If the stop position was reached, writes the Stopped state via setStopPositionStateImmediate on the same worker connection, inside the same transaction — so position, row changes, and stop state commit atomically and no cross-connection lock ordering exists
3. Commits the worker's MySQL transaction
4. Briefly takes serialMu to update vplayer bookkeeping (recordPositionSave, pending FIELD-refresh counters)
5. Calls markCommitted() to release the transaction's inflight state in the scheduler
6. Sends on txn.done (a buffered channel) to let the worker reuse the connection; returns io.EOF if the stop position was reached

Startup and Configuration

Flag: --vreplication-parallel-replication-workers N

• Default: 1 (serial apply, no parallelism)
• Set to N > 1 to enable parallel apply with N worker goroutines; capped at 64 (the per-workflow override rejects larger values, the tablet flag clamps with a warning)
• Each worker holds two MySQL connections (double-buffered apply), so a workflow uses 2N+2 connections total
• Can also be set per-workflow via the vreplication-parallel-replication-workers config override in the workflow's options column

Activation path:

1. vreplicator.replicate() creates a vplayer and calls vp.play() → vp.fetchAndApply()
2. fetchAndApply checks ParallelReplicationWorkers > 1 and len(copyState) == 0
3. If both conditions are met, calls vp.applyEventsParallel() instead of vp.applyEvents()
4. The parallel applier creates N applyWorker instances, each with its own filtered vdbClient MySQL connection
5. FK constraints are queried from information_schema at startup

Parallel apply is only active during the replication (running) phase, not during the copy phase. During copy, the serial applier is always used.

Shutdown Protocol

applyEventsParallel orchestrates a clean shutdown sequence:

1. scheduleLoop returns (either from error, context cancellation, or relay log EOF)
2. scheduler.close() is called — broadcasts to wake all blocked workers
3. wg.Wait() — waits for all worker goroutines to exit
4. close(commitCh) — signals commitLoop that no more transactions will arrive
5. <-commitDone — waits for commitLoop to drain remaining buffered transactions
6. Final error is determined: applyErr (from commitLoop/scheduleLoop) takes priority over workerErr
7. io.EOF and context.Canceled are converted to nil (the caller treats nil as a clean stop)

Relationship to MySQL Multi-Threaded Applier

MySQL's multi-threaded replica applier (MTA) has three dependency tracking modes:

The Vitess parallel applier most closely resembles WRITESET mode, but with key differences:

1. Writesets are computed on the target, not taken from the source binlog. This means parallelism is available regardless of the source's binlog_transaction_dependency_tracking setting.

2. Commit metadata is used as a fallback, not as the primary mechanism. When a writeset can be computed (non-empty PK indices, no errors), writeset-only conflict detection is used. The commitParent field is only used when the writeset is empty (build failure or no row events).

3. FK awareness is built in. MySQL's MTA does not consider FK constraints in its writeset tracking. The Vitess parallel applier queries information_schema.KEY_COLUMN_USAGE at startup and generates additional writeset keys that create conflicts between child and parent table operations.

4. Commit ordering is strict. Unlike MySQL's MTA which can commit out of order in WRITESET mode (and reorder via the slave_preserve_commit_order setting), the Vitess parallel applier always commits in order. This simplifies position tracking (single position in _vt.vreplication) and avoids gaps in the position that could confuse WaitForPos or other external observers.

Key Design Trade-offs

Strict commit ordering vs. throughput

Committing in strict order means a slow transaction can stall all commits behind it, even if their MySQL work is done. We accept this because:

• Position tracking requires monotonic progress
• WaitForPos and monitoring tools expect a single consistent position
• The commitLoop itself is fast (just UPDATE _vt.vreplication + COMMIT); the bottleneck is the workers' apply time

Writeset-only detection vs. commit-parent chains

By ignoring the source's commit-parent chain when a valid writeset exists, we achieve parallelism even when the source uses COMMIT_ORDER. The trade-off is that we must build writesets ourselves, which adds CPU overhead in the scheduleLoop. In practice, writeset computation is cheap (xxhash digests of PK values with no heap allocations).

FK batching trade-off

Without FK constraints, transaction batching reduces MySQL COMMIT overhead. With FK constraints, batching merges independent parent/child operations into single large writesets that always conflict, destroying parallelism. The solution is to skip batching when FK refs are present, accepting more frequent COMMITs in exchange for actual parallelism.

Head-of-line blocking in the pending queue

popReadyLocked stops scanning at the first non-ready, non-noConflict transaction. This prevents dispatching a later transaction whose inflight state could block the earlier one from ever becoming ready — which would deadlock with the commitLoop's strict ordering. The trade-off is that a single blocked transaction at the head of the queue stalls all transactions behind it, even non-conflicting ones. This is a correctness-over-throughput choice.

Per-worker vplayer view

Each worker builds a narrow vplayer value once at startup via workerLocalVPlayer, containing only the fields workers are allowed to share (tablePlans + its mutex/version, replicatorPlan, serialMu, etc). This gives each worker its own dbClient/query/commit bindings without fine-grained locking, while structurally preventing workers from reaching into main-goroutine-owned vplayer state. The shared mutable fields on vplayer are pointers (*sync.Mutex, *sync.RWMutex, *atomic.Int64, *atomic.Pointer) so the view and the orchestrator's vplayer alias the same synchronization state; per-session state like foreign_key_checks lives on the vdbClient (per connection) instead of the vplayer. Note this makes "vplayer is safely copyable" a load-bearing property: new mutable non-pointer fields must not be added without considering the worker view.

Source Files

Benchmarks

Benchmark Suite

A local benchmark suite is included in examples/benchmark/ to measure parallel applier throughput in isolation. The suite consists of:

Methodology

The benchmark uses a pause-load-resume approach to isolate applier throughput from vstreamer rate:

1. Setup: Create a MoveTables workflow from commerce → customer and let the copy phase complete
2. Stop: Pause the workflow so no replication occurs
3. Index: Add ~25 secondary indexes per table on the target to create expensive index maintenance overhead during apply — this ensures each statement does meaningful I/O work
4. Backlog: Generate 200K mixed operations (50% INSERT, 20% UPDATE, 5% DELETE, 15% bulk UPDATE, 10% bulk DELETE) on the source while the workflow is stopped, using a fixed random seed for deterministic, repeatable workloads
5. Drain: Start the workflow and measure wall-clock time until the target's GTID position catches up to the source's GTID position
6. Validate: Verify row counts match between source and target for all 4 tables

Timing is GTID-based, not lag-based. The benchmark captures the source's GTID sequence number after generating the backlog, then polls the target's GTID sequence until it reaches or exceeds the source. This provides an exact measurement of when all backlog transactions have been applied.

Configuration

The benchmark tunes MySQL to remove fsync as a variable and isolate the applier:

• 32MB InnoDB buffer pool on target tablets (with innodb_buffer_pool_chunk_size=1M to allow sub-128MB sizing)
• innodb_flush_log_at_trx_commit=0, sync_binlog=0 on all tablets
• innodb_change_buffering=none on target tablets (forces immediate B-tree page reads on every DML)
• durability-policy=none (no semi-sync)
• Default relay log size (250KB / 5000 items)
• VReplication experimental flags=13 (OptimizeInserts + VPlayerBatching + AllowNoBlobBinlogRowImage)

Results

Row count validation: PASS (all 4 tables match source ↔ target).

The numbers above reflect a representative single A/B run on the same hardware. Across 10+ iterations of the same benchmark, individual drain times vary (parallel: 283–330s, 606–706 ops/sec; serial: 419–521s, 383–477 ops/sec), with single-run speedup ratios ranging from ~1.4x to 1.7x. The parallel applier consistently hits 600+ ops/sec and 280–330s drain; serial throughput is more sensitive to OS-level page cache and disk I/O variance between runs, which is the dominant source of speedup-ratio variance.

Key Findings

• The relay log size is critical. Large relay logs (250MB / 500K items) let the serial applier build massive mega-transactions — merging all ~194K source transactions into a single MySQL transaction with one COMMIT. This amortizes per-commit overhead so effectively that the serial applier outperforms the parallel applier. Default relay log sizes (250KB / 5000 items) limit serial batches to ~200 source transactions per commit, which exposes the applier bottleneck and allows parallelism to help.

• Buffer pool sizing matters for parallel workers. With an 8MB buffer pool and 4 workers, each worker effectively has 2MB of InnoDB buffer pool, causing destructive cache thrashing between workers. Sizing the buffer pool so each worker gets at least as much cache as the serial applier gets total (32MB / 4 = 8MB per worker) eliminates this problem.

• FIELD event handling affects parallelism. FIELD events (table metadata) previously fell through to the default case in scheduleItems, setting curRowOnly=false and causing ~50% of transactions to be forceGlobal=true — effectively serializing the scheduler. Adding an explicit FIELD event handler reduced forceGlobal to ~1.8% (only at startup when table plans haven't been applied yet).

• Transaction batching (maxBatchedCommits) reduces per-commit overhead. Setting maxBatchedCommits = workerCount * 4 (=16 for 4 workers) merges multiple source transactions into each mega-transaction, reducing the number of MySQL COMMITs + position updates + scheduler dispatches by 16x while still producing enough independent mega-transactions to keep all workers busy.

Related Issue(s)

Checklist

• "Backport to:" labels have been added if this change should be back-ported to release branches
• If this change is to be back-ported to previous releases, a justification is included in the PR description
• Tests were added or are not required
• Did the new or modified tests pass consistently locally and on CI?
• Documentation was added or is not required

AI Disclosure

I worked with OpenCode + Codex-5.2 and Opus 4.6 + Copilot on this.